package com.swak.frame.server;

import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Sets;
import com.swak.frame.cache.CacheProxy;
import com.swak.frame.common.BooleanEnum;
import com.swak.frame.util.GetterUtil;
import com.swak.frame.util.IPHelper;
import com.swak.frame.util.UUIDHexGenerator;

/**
 * 缓存服务注册
 * ClassName: ServerRegisterServiceImpl.java
 * 
 * @author colley.ma
 * @date 2021年3月23日 下午2:33:04
 */
public class ServerRegisterServiceImpl implements ServerRegisterService {

    protected Logger logger = LoggerFactory.getLogger(getClass()); // 日志记录

    private CacheProxy cacheProxy;


    private ServerRegisterConfig serverconfig;

    private AtomicInteger instances = new AtomicInteger(0);


    @Override
    public void loadResetRegister() {
        Set<String> reloadserverSet = Sets.newHashSet();
        Set<String> serverList = getServerList();
        if (CollectionUtils.isNotEmpty(serverList)) {
            reloadserverSet.addAll(serverList);
        }
        String cacheKey = cacheProxy.format(ServerConstant.RELOAD_SERVER_LIST);
        Map<String, String> reloadServerList = cacheProxy.hGetAllString(cacheKey);
        if (MapUtils.isNotEmpty(reloadServerList)) {
            reloadserverSet.addAll(reloadServerList.keySet());
        }
        logger.info("loadReset server list:{}", JSON.toJSONString(reloadserverSet));
        // 同步到刷新ip列表
        reloadserverSet.forEach(serverIp -> {
            cacheProxy.hSet(cacheKey, serverIp, BooleanEnum.NO.get().toString());
        });
    }

    @Override
    public boolean checkCompleted() {
        String cacheKey = cacheProxy.format(ServerConstant.RELOAD_SERVER_LIST);
        Map<String, String> reloadServerList = cacheProxy.hGetAllString(cacheKey);
        if (MapUtils.isEmpty(reloadServerList)) {
            return false;
        }

        logger.info("reloadServerList:{}", JSON.toJSONString(reloadServerList));
        int needRefreshCount = 0;
        for (Entry<String, String> entry : reloadServerList.entrySet()) {
            if (GetterUtil.getLong(entry.getValue()) <= 0) {
                needRefreshCount = needRefreshCount + 1;
            }
        }
        // needRefresh==0 说明全部已经刷新了
        return needRefreshCount == 0;
    }

    @Override
    public Set<String> getServerList() {
        String cacheKey = cacheProxy.format(ServerConstant.SERVER_LIST);
        Map<String, String> serverList = cacheProxy.hGetAllString(cacheKey);
        if (MapUtils.isEmpty(serverList)) {
            return Sets.newHashSet();
        }
        return Sets.newHashSet(serverList.keySet());
    }

    @Override
    public void unRegister(String serverIp) {
        String cacheKey = cacheProxy.format(ServerConstant.SERVER_LIST);
        cacheProxy.hDel(cacheKey, serverIp);
    }

    @Override
    public void heatbeart() {
        String lockKey = cacheProxy.format(ServerConstant.LOCK_SERVER_LIST_HEATBEART);
        // 拿到锁才执行,一次只能一机器执行
        if (cacheProxy.set(lockKey, UUIDHexGenerator.generator(), serverconfig.getHeatDelay(),
            TimeUnit.MINUTES, false)) {
            logger.info("heatbeart监控,ip:{} 执行中, >>>>>>>>>>>>>>>>>>>>>>>>>>", IPHelper.localIp());
            String cacheKey = cacheProxy.format(ServerConstant.RELOAD_SERVER_LIST);
            Set<String> serverList = heatbeartServer();
            Map<String, String> reloadServerList = cacheProxy.hGetAllString(cacheKey);
            logger.warn("reloadServerList:{}", JSON.toJSONString(reloadServerList));
            Iterator<Entry<String, String>> iterator = reloadServerList.entrySet().iterator();
            Set<String> notAliveServer = Sets.newHashSet();
            while (iterator.hasNext()) {
                Entry<String, String> entry = iterator.next();
                if (!serverList.contains(entry.getKey())) {
                    logger.warn("Server ip:{},不存活移除", entry.getKey());
                    notAliveServer.add(entry.getKey());
                    iterator.remove();
                }
            }
            if (CollectionUtils.isNotEmpty(notAliveServer)) {
                // 移除不存活的iP
                cacheProxy.hDel(cacheKey, notAliveServer);
            }
            // 同步到刷新ip列表
            serverList.forEach(serverIp -> {
                if (!reloadServerList.containsKey(serverIp)) {
                    // 第一次注册到待刷新列表
                    if (!cacheProxy.hExists(cacheKey, serverIp)) {
                        cacheProxy.hSet(cacheKey, serverIp, BooleanEnum.YES.get().toString());
                    }
                }
            });
        } else {
            logger.info("heatbeart监控,其它机器执行中,跳过存活检测,ip:{} >>>>>>>>>>>>>>>>>>>>>>>>>>",
                IPHelper.localIp());
        }
    }

    public Set<String> heatbeartServer() {
        String cacheKey = cacheProxy.format(ServerConstant.SERVER_LIST);
        Map<String, Long> serverList = cacheProxy.hGetAllIncr(cacheKey);
        logger.warn("serverList:{}", JSON.toJSONString(serverList));
        if (MapUtils.isEmpty(serverList)) {
            return Sets.newHashSet();
        }
        Iterator<Entry<String, Long>> iterator = serverList.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, Long> entry = iterator.next();
            // 大于heatPeriod 默认5分钟
            if ((System.currentTimeMillis() - entry.getValue()) > serverconfig.getHeatPeriod()) {
                unRegister(entry.getKey());
                logger.warn("Server list ip:{},剩余时间:{} ms,不存活移除", entry.getKey(),
                    (System.currentTimeMillis() - entry.getValue()));
                iterator.remove();
            }
        }
        return Sets.newHashSet(serverList.keySet());
    }

    @Override
    public void serverRegister() {
        String serverList = cacheProxy.format(ServerConstant.SERVER_LIST);
        String reloadServerlist = cacheProxy.format(ServerConstant.RELOAD_SERVER_LIST);
        String localIp = IPHelper.localIp();
        cacheProxy.hSet(serverList, localIp, String.valueOf(System.currentTimeMillis()));
        cacheProxy.hSet(reloadServerlist, localIp, BooleanEnum.YES.get().toString());
        instances.lazySet(getServerList().size());
    }

    @Override
    public int getInstances() {
        return instances.get();
    }

    public void setCacheProxy(CacheProxy cacheProxy) {
        this.cacheProxy = cacheProxy;
    }

    public void setServerconfig(ServerRegisterConfig serverconfig) {
        this.serverconfig = serverconfig;
    }

    @Override
    public ServerRegisterConfig getServerconfig() {
        return serverconfig;
    }
}
